const ipc = require('node-ipc');
const log4js = require('log4js');
const path = require('path');
log4js.configure({
    appenders: { 'ipc': { type: 'dateFile', daysToKeep: 15, layout: {
        type: 'pattern',
        pattern: '[%d{yyyy-MM-dd hh:mm:ss}] [%p] %c - %m'
    }, compress: false, filename: path.join(process.cwd(), 'logs', 'ipc.log') },
    stat: { type: 'dateFile', daysToKeep: 2, layout: {
        type: 'pattern',
        pattern: '[%d{yyyy-MM-dd hh:mm:ss}] [%p] %c - %m'
    }, compress: false, filename: path.join(process.cwd(), 'logs', 'stat.log') }},
    categories: { default: { appenders: ['ipc'], level: 'info' },stat:  { appenders: ['stat'], level: 'info' }},
    pm2: true,
    pm2InstanceVar: 'INSTANCE_ID'
});
const logger = log4js.getLogger('ipc');
const loggerStat = log4js.getLogger('stat');
const config = require('./config');
const Queue = require('better-queue');
const BetterQueueLokis = require('./better-queue-lokis');
const store = new BetterQueueLokis();
const DataProcess = require('./service/DataProcess');

const q = new Queue(function (inputs, cb) {
    let length = 0;
    inputs.forEach(async (input, index) => {
        let ret;
        if (input.type === 'insert') {
            ret = await DataProcess.saveBatch(input.sqlData, input.dbInfo);
            logger.info('save task ret: ', ret);
        } else {
            ret = await DataProcess.updateBatch(input.sqlData, input.dbInfo);
            logger.info('update task ret: ', ret);
        }
        if (ret === -1) {
            this.failedTask(index, input.taskId);
        } else {
            this.finishTask(index, input.taskId);
        }
        length ++;
        if (length === inputs.length) {
            cb(null, 'success');
        }
    });
}, {
    filo: true,
    batchSize: config.batchSize || 10,
    batchDelay: 0,
    batchDelayTimeout: 5000,
    store
});

q.on('task_finish', (_task_id, taskId) => {
    const socket = sockets.get(taskId);
    if (socket) {
        ipc.server.emit(socket, 'finish_req', {
            taskId,
            status: 'finish'
        });
        delete sockets[taskId];
    }
});

q.on('task_failed', (_task_id, taskId) => {
    const socket = sockets.get(taskId);
    if (socket) {
        ipc.server.emit(socket, 'finish_req', {
            taskId,
            status: 'failed'
        });
        delete sockets[taskId];
    }
});

setInterval(() => {
    const stats = q.getStats();
    loggerStat.info(stats);
}, 500);

ipc.config.id   = 'crate';
ipc.config.retry= 1500;
ipc.config.logger = function(){};
const sockets = new Map();

ipc.serve(function(){
    ipc.server.on('insert_data', (params, socket) => {
        const taskId = params.taskId;
        sockets.set(taskId, socket);
        q.push(params);
        // logger.info('add task, ', taskId);
    });
    ipc.server.on('update_data', (params, socket) => {
        const taskId = params.taskId;
        sockets.set(taskId, socket);
        q.push(params);
        // logger.info('add update task, ', taskId);
    });
    ipc.server.on(
        'socket.disconnected',
        function(socket, destroyedSocketID) {
            ipc.log('client ' + destroyedSocketID + ' has disconnected!');
        }
    );
});

ipc.server.start();